package server

import (
	"crypto/sha1"
	"github.com/panjf2000/ants/v2"
	"github.com/xtaci/kcp-go/v5"
	"golang.org/x/crypto/pbkdf2"
	"time"
	"wths/wthsjy-im/plugins/log"
	"wths/wthsjy-im/protocol"
	"wths/wthsjy-im/server/client"
)

func kcpBlockCrypt() (kcp.BlockCrypt, error) {
	key := pbkdf2.Key([]byte(kcpKeyPassword), []byte(kcpKeySalt), 1024, 32, sha1.New)
	return kcp.NewAESBlockCrypt(key)
}

// runKCP kcp[base udp]
func (ims *ImServer) runKCP() error {
	block, _ := kcpBlockCrypt()
	listener, err := kcp.ListenWithOptions(ims.conf.Port, block, 10, 3)
	if err != nil {
		log.Error("runKCP").Msgf("%v", err)
		return err
	}
	for {
		conn, err := listener.AcceptKCP()
		if err != nil {
			log.Error("runKCP").Msgf("%v", err)
		}
		conn.SetNoDelay(1, 10, 2, 1)
		conn.SetStreamMode(true)
		conn.SetWindowSize(kcpSndwnd, kcpRcvwnd)
		_ = conn.SetReadBuffer(readBuffer)
		_ = conn.SetWriteBuffer(writeBuffer)
		conn.SetACKNoDelay(true)
		go ims.handleKcpSession(conn)
	}
}

func (ims *ImServer) handleKcpSession(conn *kcp.UDPSession) {
	cli := client.NewClient(conn, ims.conf.HandlerFunc, ims.topicMgr)
	defer cli.Close()

	for {
		_ = conn.SetReadDeadline(time.Now().Add(time.Second * 60))
		p, err := protocol.ReadIO(conn)
		if err != nil {
			log.Error("handleKcpSession").Msg(err)
			return
		}
		if err = ants.Submit(func() {
			cli.HandleProtocol(p)
		}); err != nil {
			log.Error("handleKcpSession").Msg(err)
			return
		}
	}
}
